package com.doitedu.core

import com.doitedu.beans.Student
import com.doitedu.utils.{MyPartitioner, SparkUtil}
import org.apache.spark.rdd.RDD

/**
 * @Date: 22.7.1 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C14_Transformation {
  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSparkContext("转换算子")
    val rdd1 = sc.parallelize(List(1,1,1,2,3,2,2,3,3,1), 4)
    val rdd2 = sc.parallelize(Array(4,4, 5, 6, 7, 8), 2)
    val rdd3 = sc.parallelize(Array("java" , "scala" ,"sql"), 2)

 /*   val kvRDD  = rdd3.map(word => (word.toUpperCase(), word))
    println(kvRDD.getNumPartitions)*/

    /**
     * groupByKey  只有KV的数据才能调用这个方法
     *   kvRDD.groupByKey()
     */

    val stuRDD = sc.textFile("data/stu")

    /**
     * stuRDD 原始RDD两个分区
     * 数据分布
     * */
    stuRDD.mapPartitionsWithIndex((p, iter) => {
      iter.map(stu => {
        (stu, "分区号是: " + p)
      })
    }).foreach(println)

    println("=============map方法不会改变分区内的数据分布=====================")

    val kvRDD: RDD[(String, Student)] = stuRDD.map(line => {
      val arr = line.split(",")
      (arr(0), Student(arr(0).toInt, arr(1), arr(2), arr(3).toDouble))
    })

    /**
     * kvRDD数据分布
     */

    val partitionNUmberRDD = kvRDD.mapPartitionsWithIndex((p, iter) => {
      iter.map(tp => {
        (tp._1, tp._2, "分区号是: " + p)
      })
    })

    /**
     * Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
    (1,Student(1,雨哥,语文,4.0),分区号是: 0)
    (3,Student(3,,数学,93.5),分区号是: 1)
    (1,Student(1,雨哥,数学,3.0),分区号是: 0)
    (3,Student(3,  ,英语,90.5),分区号是: 1)
    (1,Student(1,雨哥,英语,0.0),分区号是: 0)
    (4,Student(4,波波哥,语文,99.5),分区号是: 1)
    (2,Student(2,昊哥,语文,94.0),分区号是: 0)
    (4,Student(4,波波哥,数学,99.5),分区号是: 1)
    (2,Student(2,昊哥,数学,93.0),分区号是: 0)
    (4,Student(4,波波哥,英语,99.5),分区号是: 1)
    (2,Student(2,昊哥,英语,90.0),分区号是: 0)
    (3,Student(3,行哥,语文,94.5),分区号是: 0)
     */
    partitionNUmberRDD.foreach(println)

    /**
     * 按照学生的ID重新分组
     */
    val keyedRDD: RDD[(String, Iterable[Student])] = kvRDD.groupByKey(2)
    println(keyedRDD.getNumPartitions)


    /**
     * 相同id的数据一定在一组
     * 根据key分组后的数据的分布
     *  shuffle
     *
     */
    println("=============分组后各个分区内的数据分布=====================")
    keyedRDD.mapPartitionsWithIndex((p,iter)=>{
      iter.flatMap(tp=>{
        tp._2.map(stu=>(tp._1 , stu , "分区号是:"+p))
      })
    }).foreach(println)
    println("==========================自定义分区=================")

    // 将key传入到分区器中  调用 getPartition(key: Any):
    val partitioned: RDD[(String, Iterable[Student])] = kvRDD.groupByKey(new MyPartitioner(2))
    partitioned.mapPartitionsWithIndex((p,iter)=>{
      iter.flatMap(tp=>{
        tp._2.map(stu=>(tp._1 , stu , "分区号是:"+p))
      })
    }).foreach(println)
    sc.stop()
  }
  /**
   * distinct  去重
   * 1 默认分区数据不变
   * 2 数据去重后 数据量如果大大的较少  ,适当的减少分区数
   *     coalesce(2)
   **/
  def testDistinct(rdd1: RDD[Int]): Unit ={


    val resRDD = rdd1.distinct()
    // 减少分区    调用减少分区的方法
    val res = resRDD.coalesce(2)
    println(res.getNumPartitions)
  }

  /**
   * intersection
   * 1 两个RDD数据的交集 在两个RDD中都出现的元素   去重
   * 2 结果RDD的分区数  RDD分区数多的哪个RDD的分区数
   */
  private def testIntersect(rdd1: RDD[Int], rdd2: RDD[Int]): Unit = {

    val resRDD = rdd1.intersection(rdd2)
    // 获取结果RDD的分区数
    println(resRDD.getNumPartitions)
  }

  /**
   * union 算子
   * 1 两个RDD处理数据的并集  不会去重
   * 2 要求两个RDD的处理的数据类型一致 否则报错
   * 3 rdd1和rdd2的分区个数可以不一致
   * 4 返回的结果RDD的分区数是  两个RDD分区数的   和
   */
  private def testUnion(rdd1: RDD[Int], rdd2: RDD[Int]): Unit = {
    val resRDD = rdd1.union(rdd2)
    println(resRDD.getNumPartitions)


  }
}
